package cn.linjianhui.demo.flink.windows;

import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class OneSecondIntervalTrigger extends Trigger<Object, TimeWindow> {

    private ValueState<Boolean> firstSeen;

    @Override
    public TriggerResult onElement(Object o, long l, TimeWindow timeWindow, TriggerContext tc) throws Exception {
        ValueStateDescriptor<Boolean> stateDescriptor = new ValueStateDescriptor("firstSeen", Boolean.class);
        firstSeen = tc.getPartitionedState(stateDescriptor);

        if(!firstSeen.value()) {
            long t = tc.getCurrentWatermark() + (1000 - tc.getCurrentWatermark() % 1000);
            tc.registerEventTimeTimer(t);
        }

        return null;
    }

    @Override
    public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        return null;
    }

    @Override
    public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        return null;
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {

    }
}
